package com.atguigu.flink.sql.connector;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * Created by Smexy on 2023/4/14
 */
public class Demo7_UpsertKafka
{
    public static void main(String[] args) {

        //1.获取一个源表
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        SingleOutputStreamOperator<WaterSensor> ds = env
            .socketTextStream("hadoop102", 8888)
            .map(new WaterSensorMapFunction());

        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env);
        tableEnvironment.createTemporaryView("t2",ds);

        /*
              sink.partitioner: 数据写入kafka如何分区
         */
        String createTableSql = " create table t1 ( id STRING, sumVc DOUBLE , " +
            //必须加
            "   PRIMARY KEY (id) NOT ENFORCED   )with(" +
            "  'connector' = 'upsert-kafka', " +
            "  'topic' = 'topicF'," +
            "  'properties.bootstrap.servers' = 'hadoop102:9092'," +
            "  'key.format' = 'json' ," +
            "  'value.format' = 'json' ) " ;

        //建表  目标表需要映射到文件系统，目标表中的数据是写入到文件系统
        tableEnvironment.executeSql(createTableSql);

        /*
            写出
            使用 普通的 kafka连接器，要求表中的数据，只能是 insert操作。

            现在有聚合，当前表中的数据既有
                +I  Insert
                还有
                -U
                +U

                如果有这种数据，需要修改 kafka中已经存储的结果，但是由于kafka只能append，不支持修改。

                需要使用一种特殊的机制，来通过append的方式，表达 修改操作。
                需要使用upsert-kafka 连接器。

                upsert-kafka可以消费 changelog流中的数据，写入kafka。
                    +I,+U 的数据，是正常写入。以 key-value的形式写出， key是主键
                    -U的数据，会 以 key-value的形式写出， key是主键
                    -D的数据，会 以 key-value的形式写出， key是主键，value = null

                    需要生成key，key就是主键，相同key的数据放入同一个分区。



         */
        tableEnvironment.executeSql(" insert into t1 select id,sum(vc) sumVc from t2 group by id ");

    }
}
